package com.kafka.kafkademo;

import com.alibaba.fastjson.JSON;
import com.kafka.kafkademo.dto.Order;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;


/**
 * @author liyanhao
 * @Date 2023-02-18 15:03
 */
public class MsgProducer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.139.30:9092");
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);//消费重试次数
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3);//重试间隔时间
        //设置发送消息的本地缓冲区 如果设置了改缓冲区 消息会先发送到本地缓冲区可以提高消息发送性能，默认32m
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        //kafka本地线程会从缓冲区取数据 批量发送到broker
        //设置批量发送消息的大小  默认16384即16k，就是说一个batch满了16kb就发送出去
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        //默认值是0 意思就是消息必须立即被发送，但这样会影响性能
        //一般设置10毫秒左右，就是说这个消息发送完之后会进入本地的一个batch，如果10毫秒内 这个batch满了16kb就会岁batch一起被发送出去
        //如果10毫秒内 batch没满 那么也必须把消息发送出去 不能让消息的发送延迟时间太久
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        //把发送的key从字符串序列化为字节数组
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数据
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        int msgNum = 5;
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        for (int i = 0; i <= msgNum; i++) {
            Order order = new Order(i,100+i,1,1000);
            //指定发送分区 具体的发送分区计算公式 hash(key)%partitionNum
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME,
                    0,order.getI()+"", JSON.toJSONString(order));
            //未指定发送分区 具体发送的分区计算公式 hash(key)%partitionNum
            ProducerRecord<String, String> producerRecord1 = new ProducerRecord<>
                    (TOPIC_NAME, order.getI() + "", JSON.toJSONString(order));
            //等待消息发送成功的同步阻塞方法
            RecordMetadata recordMetadata = producer.send(producerRecord).get();
            System.out.println("同步方式发送消息结果:" + "topic-" + recordMetadata.topic() + "|partition-" + recordMetadata.partition() +
                    "|offset-" + recordMetadata.offset());
            //异步回调方式发送消息
            producer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e != null) {
                        System.out.println("发送消息失败：" + e.getStackTrace());
                    }
                }
            });
            if (recordMetadata != null) {
                System.out.println("异步方式发送消息结果:" + "topic-" + recordMetadata.topic() + "|patition-" + recordMetadata.partition() +
                        "|offset-" + recordMetadata.offset());
            }
            countDownLatch.countDown();
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.close();
    }
}



























